package com.iwhalecloud.uncc.dao.impl;

import java.io.StringReader;
import java.lang.reflect.Method;
import java.text.NumberFormat;
import java.text.ParseException;
import java.util.*;
import java.util.concurrent.TimeUnit;

import com.iwhalecloud.uncc.common.PersistConstant;
import org.apache.commons.lang.exception.ExceptionUtils;
import org.springframework.dao.DataAccessException;
import org.springframework.data.redis.connection.*;
import org.springframework.data.redis.connection.jedis.JedisConnectionFactory;
import org.springframework.data.redis.core.RedisCallback;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.types.Expiration;
import org.springframework.data.redis.serializer.RedisSerializer;

import com.iwhalecloud.uncc.dao.RedisClusterDao;
import com.iwhalecloud.uncc.model.RedisConfig;
import com.iwhalecloud.uncc.utils.Byte2ObjectUtil;
import com.iwhalecloud.uncc.utils.StringUtil;

import lombok.extern.slf4j.Slf4j;
import redis.clients.jedis.JedisPoolConfig;

@Slf4j
public class RedisClusterDaoImpl implements RedisClusterDao {

	private static final int EXPIRE_DAY = 7;

	private RedisTemplate<?, ?> redisTemplate;
	// 生存时间60秒
	private static final long DEFAULT_EXPIRE_TIME = 60;
	@Override
	public void init(RedisConfig redisConfig) {
		try {
			log.info("************redis cluster template load start************");

			String[] hostPorts = redisConfig.getHostPort().split(",");
			String userName = redisConfig.getUsername();
			String password = redisConfig.getPwd();
			int database = redisConfig.getDb();
			long period = redisConfig.getPeriod();
			int monitorTimeout = redisConfig.getMonitorTimeout();
			String poolProperties = redisConfig.getPoolProperties().replaceAll("@@", "\n");

			Set<RedisNode> redisNodes = new HashSet<RedisNode>();
			for (String str : hostPorts) {
				String host = str.split(":")[0];
				Integer port = Integer.parseInt(str.split(":")[1]);
				RedisNode redisNode = new RedisNode(host, port);
				redisNodes.add(redisNode);
			}

			RedisClusterConfiguration redisClusterConfiguration = new RedisClusterConfiguration();
			redisClusterConfiguration.setClusterNodes(redisNodes);
			if (password != null && !"".equals(password)) {
				redisClusterConfiguration.setPassword(RedisPassword.of(password));
			}

			JedisPoolConfig jedisPoolConfig = new JedisPoolConfig();
			final Properties pp = new Properties();
			pp.load(new StringReader(poolProperties));
			for (Enumeration<Object> enumeration = pp.keys(); enumeration.hasMoreElements();) {
				String key = enumeration.nextElement().toString();
				Method[] methods = JedisPoolConfig.class.getMethods();
				String methodName = "set" + key.substring(0, 1).toUpperCase() + key.substring(1);
				for (Method m : methods) {
					if (methodName.equals(m.getName())) {
						Object value = StringUtil.castStringToX(m.getParameterTypes()[0], pp.getProperty(key));
						m.invoke(jedisPoolConfig, value);
						break;
					}
				}
			}

			JedisConnectionFactory jedisConnectionFactory = new JedisConnectionFactory(redisClusterConfiguration,
					jedisPoolConfig);
			jedisConnectionFactory.afterPropertiesSet();

			redisTemplate = new RedisTemplate<String, Object>();
			redisTemplate.setConnectionFactory(jedisConnectionFactory);
			redisTemplate.afterPropertiesSet();

			log.info("************redis cluster template load end************");
		} catch (Exception e) {
			log.error(ExceptionUtils.getStackTrace(e));
		}

	}

	@Override
	public void set(String key, Object value) {
		redisTemplate.execute(new RedisCallback<Boolean>() {
			@Override
			public Boolean doInRedis(RedisConnection connection) throws DataAccessException {
				connection.set(key.getBytes(), Byte2ObjectUtil.objectToByte(value));
				return true;
			}
		});
	}

	@Override
	public void setEx(String key, Object value) {
		redisTemplate.execute(new RedisCallback<Boolean>() {
			@Override
			public Boolean doInRedis(RedisConnection connection) throws DataAccessException {
				connection.setEx(key.getBytes(), EXPIRE_DAY * 24 * 60 * 60, Byte2ObjectUtil.objectToByte(value));
				return true;
			}
		});
	}

	@Override
	public void setEx(String key, Object value, int seconds) {
		redisTemplate.execute(new RedisCallback<Boolean>() {
			@Override
			public Boolean doInRedis(RedisConnection connection) throws DataAccessException {
				connection.setEx(key.getBytes(), seconds, Byte2ObjectUtil.objectToByte(value));
				return true;
			}
		});
	}

	@Override
	public Object get(String key) {
		return redisTemplate.execute(new RedisCallback<Object>() {
			@Override
			public Object doInRedis(RedisConnection connection) throws DataAccessException {
				byte[] resultByte = connection.get(key.getBytes());
				if (resultByte != null) {
					Object result = Byte2ObjectUtil.byteToObject(resultByte);
					return result;
				}
				return null;
			}
		});
	}

	@Override
	public Boolean setNX(String key, Object value) {
		return redisTemplate.execute(new RedisCallback<Boolean>() {
			@Override
			public Boolean doInRedis(RedisConnection connection) throws DataAccessException {
				return connection.setNX(key.getBytes(), Byte2ObjectUtil.objectToByte(value));
			}
		});
	}

	@Override
	public Map<String, String> getHashByKey(String key) {
		return redisTemplate.execute(new RedisCallback<Map<String, String>>() {
			@Override
			public Map<String, String> doInRedis(RedisConnection connection) throws DataAccessException {
				Map<String, String> result = new HashMap<String, String>();
				RedisSerializer<String> serializer = redisTemplate.getStringSerializer();
				Map<byte[], byte[]> map = connection.hGetAll(serializer.serialize(key));
				Set<Map.Entry<byte[], byte[]>> entrys = map.entrySet();
				for (Map.Entry<byte[], byte[]> entry : entrys) {
					String key = serializer.deserialize(entry.getKey());
					String value = serializer.deserialize(entry.getValue());
					if (value != null && !"null".equals(value) && !"nil".equals(value)) {
						result.put(key, value);
					}
				}
				return result;
			}
		});
	}

	@Override
	public Boolean hSet(String key, String field, String value) {
		return redisTemplate.execute(new RedisCallback<Boolean>() {
			@Override
			public Boolean doInRedis(RedisConnection connection) throws DataAccessException {
				RedisSerializer<String> serializer = redisTemplate.getStringSerializer();
				return connection.hSet(serializer.serialize(key), serializer.serialize(field),
						serializer.serialize(value));
			}
		});
	}

	@Override
	public String hGet(String key, String field) {
		return redisTemplate.execute(new RedisCallback<String>() {
			@Override
			public String doInRedis(RedisConnection connection) throws DataAccessException {
				RedisSerializer<String> serializer = redisTemplate.getStringSerializer();
				byte[] bytes = connection.hGet(serializer.serialize(key), serializer.serialize(field));
				if (bytes == null) {
					return null;
				} else {
					return serializer.deserialize(bytes);
				}
			}
		});
	}

	@Override
	public Long hdel(String key, String field) {
	return	redisTemplate.execute(new RedisCallback<Long>() {
			@Override
			public Long doInRedis(RedisConnection connection) throws DataAccessException {
				RedisSerializer<String> serializer = redisTemplate.getStringSerializer();
				return connection.hDel(serializer.serialize(key), serializer.serialize(field));
			}
		});
	}

	@Override
	public Long del(String key) {
		return redisTemplate.execute(new RedisCallback<Long>() {
			@Override
			public Long doInRedis(RedisConnection connection) throws DataAccessException {
				return connection.del(redisTemplate.getStringSerializer().serialize(key));
			}
		});
	}

	@Override
	public void lPush(String key, String value) {
		redisTemplate.execute(new RedisCallback<Long>() {
			@Override
			public Long doInRedis(RedisConnection connection) throws DataAccessException {
				RedisSerializer<String> serializer = redisTemplate.getStringSerializer();
				return connection.lPush(serializer.serialize(key), serializer.serialize(value));
			}
		});
	}

	@Override
	public String rPop(String key) {
		return redisTemplate.execute(new RedisCallback<String>() {
			@Override
			public String doInRedis(RedisConnection connection) throws DataAccessException {
				RedisSerializer<String> serializer = redisTemplate.getStringSerializer();
				byte[] keyByte = connection.rPop(serializer.serialize(key));
				if (keyByte != null) {
					String key = serializer.deserialize(keyByte);
					if (!"nil".equals(key)) {
						return key;
					}
				}
				return null;
			}
		});
	}

	@Override
	public void rPush(String key, String value) {
		redisTemplate.execute(new RedisCallback<Long>() {
			@Override
			public Long doInRedis(RedisConnection connection) throws DataAccessException {
				RedisSerializer<String> serializer = redisTemplate.getStringSerializer();
				return connection.rPush(serializer.serialize(key), serializer.serialize(value));
			}
		});
	}

	@Override
	public String lPop(String key) {
		return redisTemplate.execute(new RedisCallback<String>() {
			@Override
			public String doInRedis(RedisConnection connection) throws DataAccessException {
				RedisSerializer<String> serializer = redisTemplate.getStringSerializer();
				byte[] keyByte = connection.lPop(serializer.serialize(key));
				if (keyByte != null) {
					String key = serializer.deserialize(keyByte);
					if (!key.equals("nil")) {
						return key;
					}
				}
				return null;
			}
		});
	}

	@Override
	public List<String> lRange(String key, long start, long end) {
		return redisTemplate.execute(new RedisCallback<List<String>>() {
			@Override
			public List<String> doInRedis(RedisConnection redisConnection) throws DataAccessException {
				RedisSerializer<String> serializer = redisTemplate.getStringSerializer();
				List<byte[]> keyByte = redisConnection.lRange(key.getBytes(), start, end);
				if (keyByte == null || keyByte.isEmpty()) {
					return new ArrayList<>();
				}
				List<String> res = new ArrayList<>();
				for (byte[] bytes : keyByte) {
					String str = serializer.deserialize(bytes);
					res.add(str);
				}
				return res;
			}
		});
	}

	@Override
	public Long lLen(String key) {
		return redisTemplate.execute(new RedisCallback<Long>() {
			@Override
			public Long doInRedis(RedisConnection connection) throws DataAccessException {
				RedisSerializer<String> serializer = redisTemplate.getStringSerializer();
				return connection.lLen(serializer.serialize(key));
			}
		});
	}

	@Override
	public Long sRem(String key, String value) {
		return redisTemplate.execute(new RedisCallback<Long>() {
			@Override
			public Long doInRedis(RedisConnection connection) throws DataAccessException {
				RedisSerializer<String> serializer = redisTemplate.getStringSerializer();
				return connection.sRem(serializer.serialize(key), serializer.serialize(value));
			}
		});
	}

	@Override
	public Long incr(String key) {
		return redisTemplate.execute(new RedisCallback<Long>() {
			@Override
			public Long doInRedis(RedisConnection connection) throws DataAccessException {
				RedisSerializer<String> serializer = redisTemplate.getStringSerializer();
				return connection.incr(serializer.serialize(key));
			}
		});
	}

	@Override
	public boolean exists(String key) {
		return redisTemplate.execute(new RedisCallback<Boolean>() {
			@Override
			public Boolean doInRedis(RedisConnection connection) throws DataAccessException {
				RedisSerializer<String> serializer = redisTemplate.getStringSerializer();
				return connection.exists(serializer.serialize(key));
			}
		});
	}

	@Override
	public Number getNumber(String key) {
		return redisTemplate.execute(new RedisCallback<Number>() {
			@Override
			public Number doInRedis(RedisConnection connection) throws DataAccessException {
				byte[] resultByte = connection.get(key.getBytes());
				if (resultByte != null) {
					String valueStr = new String(resultByte);
					try {
						return NumberFormat.getInstance().parse(valueStr);
					} catch (ParseException e) {
						log.error(ExceptionUtils.getStackTrace(e));
					}
				}
				return null;
			}
		});
	}

	@Override
	public void setExHasExpire(String key, Object value, long expireTime) {
		redisTemplate.execute(new RedisCallback<Boolean>() {
			@Override
			public Boolean doInRedis(RedisConnection connection) throws DataAccessException {
				connection.setEx(key.getBytes(), expireTime, Byte2ObjectUtil.objectToByte(value));
				return true;
			}
		});
	}


	/**
	 * 获取共享锁(带超时时间)
	 *
	 * @return 1:获取锁成功,0获取锁失败
	 */
	private long acquireLock(String lockName, final long expire) {
		final String rKey =  lockName;
		Boolean object = false;
		object = redisTemplate.execute(new RedisCallback<Boolean>() {
			@Override
			public Boolean doInRedis(RedisConnection connection) {
				byte[] lockBytes = rKey.getBytes();
				Boolean locked = connection.setNX(lockBytes, lockBytes);
				connection.expire(lockBytes, expire);
				return locked;
			}
		});
		return (object != null && object) ? 1L : 0L;
	}

	/**
	 * 如果锁空闲立即返回,获取失败 一直等待
	 *
	 * @param lock
	 * @throws InterruptedException
	 */
	@Override
	public void lock(String lock)  {
		String lockName = PersistConstant.PREFIX_SYS+PersistConstant.PREFIX_LOCK+  lock;
		while (true) {
			if (log.isDebugEnabled()) {
				log.debug("lock key: " + lockName);
			}
			long i = this.acquireLock(lockName, DEFAULT_EXPIRE_TIME);
			if (i == 1) {// 获取成功
				if (log.isDebugEnabled()) {
					log.debug(
							"get lock[" + lockName + "] successfully, expire in " + DEFAULT_EXPIRE_TIME + " seconds.");
				}
				return ;
			} else {
				if (log.isDebugEnabled()) {
					log.debug(
							"[" + lockName + "] locked by another business, waiting 10 seconds will try lock again...");
				}
			}
			try {
				Thread.sleep(100);
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}
	}

	/**
	 * 释放锁
	 *
	 * @param lock
	 * @throws Exception
	 */
	@Override
	public void unLock(String lock) {
		String lockName = PersistConstant.PREFIX_SYS+PersistConstant.PREFIX_LOCK+ lock;
		del(lockName);
		if (log.isDebugEnabled()) {
			log.debug("release lock[" + lockName + "] successfully!");
		}
	}

	@Override
	public boolean setNxAndExpire(String key, Object value, long expireTime) {
		return redisTemplate.execute(new RedisCallback<Boolean>() {
			@Override
			public Boolean doInRedis(RedisConnection connection) throws DataAccessException {
				return connection.set(key.getBytes(), Byte2ObjectUtil.objectToByte(value),
						Expiration.seconds(expireTime), RedisStringCommands.SetOption.ifAbsent());
			}
		});
	}
}
